package com.kk.util.redis;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kk.config.SpringContextBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Component
public class RedisUtils {
	@Autowired
	RedisTemplate redisTemplate;

	@Autowired
	SpringContextBean springContextBean;

	private Logger log = LoggerFactory.getLogger(this.getClass());

	@Bean
	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
											MessageListenerAdapter listenerAdapter) {

		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(connectionFactory);
		// container.addMessageListener(listenerAdapter, new PatternTopic("chat"));

		return container;
	}

	@Bean
	MessageListenerAdapter listenerAdapter(Receiver receiver) {
		return new MessageListenerAdapter(receiver, "receiveMessage");
	}

	@Bean
	Receiver receiver(CountDownLatch latch) {
		return new Receiver(latch);
	}

	@Bean
	CountDownLatch latch() {
		return new CountDownLatch(1);
	}

	@Bean
	StringRedisTemplate stringTemplate(RedisConnectionFactory connectionFactory) {
		return new StringRedisTemplate(connectionFactory);
	}

	@PostConstruct
	public void init() {
		RedisSerializer<String> stringSerializer = new StringRedisSerializer();
		// redisTemplate.setKeySerializer(stringSerializer);
		// redisTemplate.setValueSerializer(stringSerializer);
		// redisTemplate.setHashKeySerializer(stringSerializer);
		// redisTemplate.setHashValueSerializer(stringSerializer);

		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
				Object.class);
		ObjectMapper om = new ObjectMapper();
		om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
		om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
		jackson2JsonRedisSerializer.setObjectMapper(om);
		redisTemplate.setKeySerializer(stringSerializer);
		redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
		redisTemplate.setHashKeySerializer(stringSerializer);
		redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
		redisTemplate.afterPropertiesSet();
	}

	public Boolean set(Object key, Object val) {
		Boolean result = false;
		try {
			redisTemplate.opsForValue().set(String.valueOf(key), val);
			result = true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return result;
	}

	public Boolean set(Object key, Object val, int time, TimeUnit unit) {
		Boolean result = false;
		try {
			redisTemplate.opsForValue().set(String.valueOf(key), val, time, unit);
			result = true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return result;
	}

	public Object get(Object key) {
		return redisTemplate.opsForValue().get(String.valueOf(key));
	}

	/**
	 * 批量删除对应的value
	 *
	 * @param keys
	 */
	public void remove(final Object... keys) {
		for (Object key : keys) {
			remove(String.valueOf(key));
		}
	}

	/**
	 * 批量删除key
	 *
	 * @param pattern
	 */
	public void removePattern(final String pattern) {
		Set<Serializable> keys = redisTemplate.keys(pattern);
		if (keys.size() > 0)
			redisTemplate.delete(keys);
	}

	/**
	 * 删除对应的value
	 *
	 * @param key
	 */
	public boolean remove(final String key) {
		boolean result = false;
		try {
			if (exists(key)) {
				redisTemplate.delete(key);
				result = true;
			} else {
				log.info("redis中不存在key:"+key);
				result = false;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return result;
	}

	/**
	 * 判断缓存中是否有对应的value
	 *
	 * @param key
	 * @return
	 */
	public boolean exists(final String key) {
		return redisTemplate.hasKey(key);
	}

	public Boolean multiSet(Map<String, Object> map) {
		Boolean result = false;
		try {
			redisTemplate.opsForValue().multiSet(map);
			result = true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return result;
	}

	public List<Object> multiGet(Collection<String> keys) {
		return redisTemplate.opsForValue().multiGet(keys);
	}

	public long incr(String key, long delta) {
		return redisTemplate.opsForValue().increment(key, delta);
	}

	public void lpush(String key, String value) {
		redisTemplate.opsForList().leftPush(key, value);
	}

	public List<Object> range(String key, int start, int end) {
		return redisTemplate.opsForList().range(key, start, end);
	}

	public Object rpop(String key) {
		return redisTemplate.opsForList().rightPop(key);
	}

	public void setHash(String key, Map<String, Object> map) {
		redisTemplate.opsForHash().putAll(key, map);
	}

	public Object getHash(String key, String prop) {
		return redisTemplate.opsForHash().get(key, prop);
	}

	public Map getHashAll(String key) {
		Map map = new HashMap();
		map.put("keys", redisTemplate.opsForHash().keys(key));
		map.put("vals", redisTemplate.opsForHash().values(key));
		return map;
	}

	public void subscribe(String channel) {
		RedisMessageListenerContainer container = springContextBean.getBean(RedisMessageListenerContainer.class);
		MessageListenerAdapter listenerAdapter = springContextBean.getBean(MessageListenerAdapter.class);
		container.addMessageListener(listenerAdapter, new PatternTopic(channel));
	}

	public void publish(String channel, String message) throws InterruptedException {
		CountDownLatch latch = springContextBean.getBean(CountDownLatch.class);

		log.info("Sending message...");
		redisTemplate.convertAndSend(channel, message);

		latch.await();
	}
}
